package com.lzq.mq.broker.event.spi.listener;

import com.alibaba.fastjson2.JSON;
import com.lzq.mq.broker.cache.CommonCache;
import com.lzq.mq.broker.event.model.PushMsgEvent;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.dto.ClientSendMessageToBrokerRespDTO;
import com.lzq.mq.common.dto.MessageDTO;
import com.lzq.mq.common.enums.BrokerMessageSendWay;
import com.lzq.mq.common.enums.BrokerResponseCode;
import com.lzq.mq.common.enums.ClientSendMessageToBrokerRespStatus;
import com.lzq.mq.common.event.Listener;

public class PushMsgListener implements Listener<PushMsgEvent> {

    @Override
    public void onReceive(PushMsgEvent event) throws Exception {
        MessageDTO messageDTO = event.getMessageDTO();
        // 写入CommitLog和ConsumeQueue
        CommonCache.getCommitLogAppendHandler()
                .appendMsg(messageDTO);
        // 返回ack
        if (messageDTO.getSendWay() == BrokerMessageSendWay.ASYNC.getCode()) {
            return;     // 异步消息不需要返回ack
        }
        ClientSendMessageToBrokerRespDTO respDTO = new ClientSendMessageToBrokerRespDTO();
        respDTO.setStatus(ClientSendMessageToBrokerRespStatus.SUCCESS.getCode());
        respDTO.setMsgId(messageDTO.getMsgId());
        TcpMsg tcpMsg = new TcpMsg(BrokerResponseCode.SEND_MSG_RESP.getCode(), JSON.toJSONBytes(respDTO));

        event.getChannelHandlerContext().writeAndFlush(tcpMsg);
    }
}
